/*
 * Copyright (c) 2024 endless-sky
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include <gst/gst.h>
#include <gst/app/gstappsink.h>
#include <gst/app/gstappsrc.h>
#include <memory>
#include <map>
#include <mutex>
#include <thread>
#include <vector>
#include <string>
#include <nlohmann/json.hpp>
#include "utils_noncopyable.h"
#include "media_stream.h"
#include "stream_flow_control.h"

namespace El {
namespace StreamService {

/**
 * @brief RTMP 推流服务（GStreamer）
 * 从本地编码数据（H.264/AAC）构建 GStreamer 管道，推送到 RTMP/RTMPS 服务器。
 */
class StreamGstRtmp {
    DISALLOW_COPY_AND_MOVE(StreamGstRtmp);

public:
    static StreamGstRtmp &GetInstance();

    bool Start();
    void Stop();

    struct PushData {
        uint32_t channel{0};
        uint32_t streamType{0};
        std::string url;

        GstElement *pipeline{nullptr};
        GstElement *appsrcVideo{nullptr};
        GstElement *appsrcAudio{nullptr};

        uint32_t videoFps{0};
        uint32_t audioSampleRate{0};
        uint32_t audioChannels{0};

        Media::StreamSourcePtr streamSource;
        int streamHandle{0};

        // 流控管理器 (使用shared_ptr,因为GstBuffer的释放回调需要延长生命周期)
        std::shared_ptr<StreamFlowControl> flowControl;

        void Cleanup()
        {
            if (streamSource) {
                streamSource->Stop();
                streamSource.reset();
            }
            if (appsrcVideo) {
                gst_object_unref(appsrcVideo);
                appsrcVideo = nullptr;
            }
            if (appsrcAudio) {
                gst_object_unref(appsrcAudio);
                appsrcAudio = nullptr;
            }
            if (pipeline) {
                gst_element_set_state(pipeline, GST_STATE_NULL);
                gst_object_unref(pipeline);
                pipeline = nullptr;
            }
            flowControl.reset();
        }
    };

    using PushDataPtr = std::shared_ptr<PushData>;

private:
    StreamGstRtmp() = default;
    ~StreamGstRtmp();

private:
    bool SetupSink(const std::string &url, uint32_t channel, uint32_t streamType);
    void PushFrame(const Media::MediaFramePtr &frame, PushDataPtr data);
    bool ApplyConfig(const nlohmann::json &rtmpConfig);
    void ClearSinks();

private:
    std::vector<PushDataPtr> sinks_;
    std::mutex mutex_;
    std::mutex reconfigMutex_;
    int configHandle_{-1};
    bool started_{false};
    bool audioEnable_{false};
    std::string audioCodec_{"aac"};
    int audioSampleRate_{8000};
    int audioChannels_{1};
    int audioBitWidth_{16};
};

} // namespace StreamService
} // namespace El
